/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.dolphinscheduler.api.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.facebook.presto.jdbc.internal.guava.base.Strings;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.dto.MonitorDto;
import org.apache.dolphinscheduler.api.dto.WorkerGroupDto;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.MonitorService;
import org.apache.dolphinscheduler.api.service.UserGroupWorkerService;
import org.apache.dolphinscheduler.api.service.WorkerGroupService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.model.WorkerServerModel;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.*;
import org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
import org.apache.dolphinscheduler.registry.api.RegistryClient;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * worker group service impl
 */
@Service
@Slf4j
public class WorkerGroupServiceImpl extends BaseServiceImpl implements WorkerGroupService {

    @Autowired
    private WorkerGroupMapper workerGroupMapper;

    @Autowired
    private ProcessInstanceMapper processInstanceMapper;

    @Autowired
    private RegistryClient registryClient;

    @Autowired
    private EnvironmentWorkerGroupRelationMapper environmentWorkerGroupRelationMapper;

    @Autowired
    private ProcessService processService;

    @Autowired
    private ScheduleMapper scheduleMapper;
    @Autowired
    private UserGroupWorkerService userGroupWorkerService;

    @Autowired
    @Lazy
    private MonitorService monitorService;

    /**
     * create or update a worker group
     *
     * @param loginUser login user
     * @param id worker group id
     * @param name worker group name
     * @param addrList addr list
     * @param central
     * @param node
     * @param connectedList
     * @return create or update result code
     */
    @Override
    @Transactional
    public Map<String, Object> saveWorkerGroup(User loginUser, int id, String name, String addrList, String description,
                                               String otherParamsJson, String userGroupId, Integer central, Integer node, String connectedList) {
        Map<String, Object> result = new HashMap<>();
//        if (!canOperatorPermissions(loginUser, null, AuthorizationType.WORKER_GROUP, WORKER_GROUP_CREATE)) {
//            putMsg(result, Status.USER_NO_OPERATION_PERM);
//            return result;
//        }
        if (StringUtils.isEmpty(name)) {
            log.warn("Parameter name can ot be null.");
            putMsg(result, Status.NAME_NULL);
            return result;
        }
        Date now = new Date();
        WorkerGroup workerGroup = null;
        if (id != 0) {
            workerGroup = workerGroupMapper.selectById(id);
        }
        if (workerGroup == null) {
            workerGroup = new WorkerGroup();
            workerGroup.setCreateTime(now);
        }
        workerGroup.setName(name);
        workerGroup.setAddrList(addrList);
        workerGroup.setUpdateTime(now);
        workerGroup.setDescription(description);

        workerGroup.setNode(node);

        // 设置节点是否为中心节点，central为1表示中心节点
        WorkerGroup centralWG = null;
        if(central == 1){

            centralWG = workerGroupMapper.queryCentralNode();

            if(centralWG != null && id != centralWG.getId().intValue()){
                log.warn("only save one central worker group.");
                putMsg(result, Status.ONLY_WORKER_GROUP_CENTRAL);
                return result;
            }
            workerGroup.setCentral(central);
        } else {
            workerGroup.setNode(node);
            workerGroup.setCentral(central);
        }

        String oldConnectedList = workerGroup.getConnectedList();

        if (checkWorkerGroupNameExists(workerGroup) && connectedList.trim().equalsIgnoreCase(oldConnectedList)) {
            putMsg(result, Status.NAME_OR_CONNECTED_LIST_EXIST, workerGroup.getName(),workerGroup.getConnectedList());
            return result;
        }

        workerGroup.setConnectedList(connectedList);

        if (checkWorkerGroupNameExists(workerGroup)) {
            log.warn("Worker group with the same name already exists, name:{}.", workerGroup.getName());
            putMsg(result, Status.NAME_EXIST, workerGroup.getName());
            return result;
        }
        String invalidAddr = checkWorkerGroupAddrList(workerGroup);
        if (invalidAddr != null) {
            log.warn("Worker group address is invalid, invalidAddr:{}.", invalidAddr);
            putMsg(result, Status.WORKER_ADDRESS_INVALID, invalidAddr);
            return result;
        }
        handleDefaultWorkGroup(workerGroupMapper, workerGroup, loginUser, otherParamsJson);
        log.info("Worker group save complete, workerGroupName:{}.", workerGroup.getName());
        
        // 保存work与用户组的关系
        if(StringUtils.isNotBlank(userGroupId)) {
        	String[] ids = StringUtils.split(userGroupId, ",");
        	List<Integer> userGroupIds = new ArrayList<Integer>();
        	if(ids.length > 0) {
        		for(String ugId : ids) {
        			userGroupIds.add(Integer.parseInt(ugId));
        		}
        	}
        	userGroupWorkerService.saveUserGroupWorkByWorkId(workerGroup.getId(), userGroupIds);
        }
        putMsg(result, Status.SUCCESS);
        return result;
    }

    protected void handleDefaultWorkGroup(WorkerGroupMapper workerGroupMapper, WorkerGroup workerGroup, User loginUser,
                                          String otherParamsJson) {
        if (workerGroup.getId() != null) {
            workerGroupMapper.updateById(workerGroup);
        } else {
            workerGroupMapper.insert(workerGroup);
            permissionPostHandle(AuthorizationType.WORKER_GROUP, loginUser.getId(),
                    Collections.singletonList(workerGroup.getId()), log);
        }
    }

    /**
     * check worker group name exists
     *
     * @param workerGroup worker group
     * @return boolean
     */
    private boolean checkWorkerGroupNameExists(WorkerGroup workerGroup) {
        // check database
        List<WorkerGroup> workerGroupList = workerGroupMapper.queryWorkerGroupByName(workerGroup.getName());
        if (CollectionUtils.isNotEmpty(workerGroupList)) {
            // create group, the same group name exists in the database
            if (workerGroup.getId() == null) {
                return true;
            }
            // update group, the database exists with the same group name except itself
            Optional<WorkerGroup> sameNameWorkGroupOptional = workerGroupList.stream()
                    .filter(group -> !Objects.equals(group.getId(), workerGroup.getId())).findFirst();
            if (sameNameWorkGroupOptional.isPresent()) {
                return true;
            }
        }
        return false;
    }

    /**
     * check worker group addr list
     *
     * @param workerGroup worker group
     * @return boolean
     */
    private String checkWorkerGroupAddrList(WorkerGroup workerGroup) {
        if (Strings.isNullOrEmpty(workerGroup.getAddrList())) {
            return null;
        }
        Map<String, String> serverMaps = registryClient.getServerMaps(RegistryNodeType.WORKER);
        for (String addr : workerGroup.getAddrList().split(Constants.COMMA)) {
            if (!serverMaps.containsKey(addr)) {
                return addr;
            }
        }
        return null;
    }

    /**
     * query worker group paging
     *
     * @param loginUser login user
     * @param pageNo page number
     * @param pageSize page size
     * @param searchVal search value
     * @param node
     * @return worker group list page
     */
    @Override
    public Result queryAllGroupPaging(User loginUser, Integer pageNo, Integer pageSize, String searchVal, Integer node) {
        // list from index
        int fromIndex = (pageNo - 1) * pageSize;
        // list to index
        int toIndex = (pageNo - 1) * pageSize + pageSize;

        Result result = new Result();
//        List<WorkerGroup> workerGroups;
//        if (loginUser.getUserType().equals(UserType.ADMIN_USER)) {
//            workerGroups = getWorkerGroups(null);
//        } else {
//            Set<Integer> ids = resourcePermissionCheckService
//                    .userOwnedResourceIdsAcquisition(AuthorizationType.WORKER_GROUP, loginUser.getId(), log);
//            workerGroups = getWorkerGroups(ids.isEmpty() ? Collections.emptyList() : new ArrayList<>(ids));
//        }
        List<WorkerGroup> workerGroups = getWorkerGroups(true, node);
        List<WorkerGroup> resultDataList = new ArrayList<>();
        int total = 0;

        if (CollectionUtils.isNotEmpty(workerGroups)) {
            List<WorkerGroup> searchValDataList = new ArrayList<>();

            if (!StringUtils.isEmpty(searchVal)) {
                for (WorkerGroup workerGroup : workerGroups) {
                    if (workerGroup.getName().contains(searchVal)) {
                        searchValDataList.add(workerGroup);
                    }
                }
            } else {
                searchValDataList = workerGroups;
            }
            total = searchValDataList.size();
            if (fromIndex < searchValDataList.size()) {
                if (toIndex > searchValDataList.size()) {
                    toIndex = searchValDataList.size();
                }
                resultDataList = searchValDataList.subList(fromIndex, toIndex);
            }
        }
//        if(resultDataList.size() > 0) {
//        	for(WorkerGroup wg : resultDataList) {
//        		wg.setUserGroups(userGroupWorkerService.queryByWorkId(wg.getId()));
//        	}
//        }

        List<WorkerGroupDto> resultDataDtoList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(resultDataList)) {
            // 查询调度器worker数据
            Map<String, MonitorDto> workerMonitorData = queryWorkerMonitorData();

            for (WorkerGroup workerGroup : resultDataList) {
                WorkerGroupDto workerGroupDto = new WorkerGroupDto();
                try {
                    PropertyUtils.copyProperties(workerGroupDto,workerGroup);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                workerGroupDto.setUserGroups(userGroupWorkerService.queryByWorkId(workerGroup.getId()));
                String addrList = workerGroup.getAddrList();
                List<String> addrs = Arrays.asList(addrList.split(","));
                // 分组监控数据平均值
                if (CollectionUtils.isNotEmpty(addrs)) {
                    // 分组运算所需的数据
                    List<MonitorDto> workerGroupMonitorList = new ArrayList<>();
                    for (String addr : addrs) {
                        MonitorDto monitorDto = workerMonitorData.get(addr);
                        if (Objects.nonNull(monitorDto)) {
                            workerGroupMonitorList.add(monitorDto);
                        }
                    }
                    if (CollectionUtils.isNotEmpty(workerGroupMonitorList)) {
                        workerGroupDto.setCpuUsage(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToDouble(MonitorDto::getCpuUsage).average().orElse(0d))
                                .setScale(2, RoundingMode.HALF_UP).doubleValue());
                        workerGroupDto.setMemoryUsage(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToDouble(MonitorDto::getMemoryUsage).average().orElse(0d))
                                .setScale(2, RoundingMode.HALF_UP).doubleValue());
                        workerGroupDto.setReservedMemory(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToDouble(MonitorDto::getReservedMemory).average().orElse(0d))
                                .setScale(2, RoundingMode.HALF_UP).doubleValue());
                        workerGroupDto.setWorkerWaitingTaskCount(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToDouble(MonitorDto::getWorkerWaitingTaskCount).sum())
                                .setScale(2, RoundingMode.HALF_UP).intValue());

                        workerGroupDto.setCpuNum(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToInt(MonitorDto::getCpuNum).sum())
                                .setScale(2, RoundingMode.HALF_UP).intValue());
                        workerGroupDto.setMemorySize(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToLong(MonitorDto::getMemorySize).sum())
                                .setScale(2, RoundingMode.HALF_UP).longValue());
//                        workerGroupDto.setSystemTotal(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToLong(MonitorDto::getSystemTotal).sum())
//                                .setScale(2, RoundingMode.HALF_UP).longValue());
//                        workerGroupDto.setSystemUsed(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToLong(MonitorDto::getSystemUsed).sum())
//                                .setScale(2, RoundingMode.HALF_UP).longValue());
                        workerGroupDto.setDataTotal(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToLong(MonitorDto::getDataTotal).sum())
                                .setScale(2, RoundingMode.HALF_UP).longValue());
                        workerGroupDto.setDataUsed(BigDecimal.valueOf(workerGroupMonitorList.stream().mapToLong(MonitorDto::getDataUsed).sum())
                                .setScale(2, RoundingMode.HALF_UP).longValue());
                        workerGroupDto.setArchitecture(workerGroupMonitorList.stream().map(MonitorDto::getArchitecture).filter(Objects::nonNull).distinct().collect(Collectors.joining("/")));

                        // 虚拟机数量
                        workerGroupDto.setVirtualMachine((int) workerGroupMonitorList.stream().filter(it->it.getIsVirtual() == 1).count());
                        // 物理机数量
                        workerGroupDto.setPhysicalMachine((int) workerGroupMonitorList.stream().filter(it->it.getIsVirtual() == 0).count());
                    }

                }
                resultDataDtoList.add(workerGroupDto);
            }
        }
        PageInfo<WorkerGroupDto> pageInfo = new PageInfo<>(pageNo, pageSize);
        pageInfo.setTotal(total);
        pageInfo.setTotalList(resultDataDtoList);

        result.setData(pageInfo);
        putMsg(result, Status.SUCCESS);
        return result;
    }

    /**
     * query all worker group
     *
     * @param loginUser
     * @return all worker group list
     */
    @Override
    public Map<String, Object> queryAllGroup(User loginUser) {
        Map<String, Object> result = new HashMap<>();
        List<WorkerGroup> workerGroups;
        if (loginUser.getUserType().equals(UserType.ADMIN_USER)) {
            workerGroups = getWorkerGroups(null);
        } else {
            Set<Integer> ids = resourcePermissionCheckService
                    .userOwnedResourceIdsAcquisition(AuthorizationType.WORKER_GROUP, loginUser.getId(), log);
            workerGroups = getWorkerGroups(ids.isEmpty() ? Collections.emptyList() : new ArrayList<>(ids));
            workerGroups = getWorkGroupByWordIds(loginUser.getUserGroupId(), workerGroups);
        }
        if(workerGroups != null && workerGroups.size() > 0) {
        	List<String> availableWorkerGroupList = workerGroups.stream()
                    .map(WorkerGroup::getName)
                    .collect(Collectors.toList());
            result.put(Constants.DATA_LIST, availableWorkerGroupList);
        } else {
        	result.put(Constants.DATA_LIST, null);
        }
        putMsg(result, Status.SUCCESS);
        return result;
    }

    /**
     * get worker groups
     *
     * @return WorkerGroup list
     */
    private List<WorkerGroup> getWorkerGroups(List<Integer> ids) {
        // worker groups from database
        List<WorkerGroup> workerGroups;
        if (ids != null) {
            workerGroups = ids.isEmpty() ? new ArrayList<>() : workerGroupMapper.selectBatchIds(ids);
        } else {
            workerGroups = workerGroupMapper.queryAllWorkerGroup();
        }
        
        boolean containDefaultWorkerGroups = workerGroups.stream()
                .anyMatch(workerGroup -> Constants.DEFAULT_WORKER_GROUP.equals(workerGroup.getName()));
        if (!containDefaultWorkerGroups) {
            // there doesn't exist a default WorkerGroup, we will add all worker to the default worker group.
            Set<String> activeWorkerNodes = registryClient.getServerNodeSet(RegistryNodeType.WORKER);
            WorkerGroup defaultWorkerGroup = new WorkerGroup();
            defaultWorkerGroup.setName(Constants.DEFAULT_WORKER_GROUP);
            defaultWorkerGroup.setAddrList(String.join(Constants.COMMA, activeWorkerNodes));
            defaultWorkerGroup.setCreateTime(new Date());
            defaultWorkerGroup.setUpdateTime(new Date());
            defaultWorkerGroup.setSystemDefault(true);
            workerGroups.add(defaultWorkerGroup);
        }

        return workerGroups;
    }

    /**
     * get worker groups
     *
     * @param isPaging whether paging
     * @param node
     * @return WorkerGroup list
     */
    private List<WorkerGroup> getWorkerGroups(boolean isPaging, Integer node) {
        // worker groups from database
        List<WorkerGroup> workerGroups;
        if(Objects.nonNull(node) && node == 1){
            workerGroups = workerGroupMapper.queryAllNodeGroup();
        } else {
            workerGroups = workerGroupMapper.queryAllWorkerGroup();
        }
        // worker groups from zookeeper
        String workerPath = Constants.REGISTRY_DOLPHINSCHEDULER_WORKERS;
        Collection<String> workerGroupList = null;
        try {
            workerGroupList = registryClient.getChildrenKeys(workerPath);
        } catch (Exception e) {
            log.error("getWorkerGroups exception, workerPath: {}, isPaging: {}", workerPath, isPaging, e);
        }

        if (org.apache.commons.collections.CollectionUtils.isEmpty(workerGroupList)) {
            if (org.apache.commons.collections.CollectionUtils.isEmpty(workerGroups) && !isPaging) {
                WorkerGroup wg = new WorkerGroup();
                wg.setName(Constants.DEFAULT_WORKER_GROUP);
                workerGroups.add(wg);
            }
            return workerGroups;
        }

        for (String workerGroup : workerGroupList) {
            String workerGroupPath = workerPath + Constants.SINGLE_SLASH + workerGroup;
            Collection<String> childrenNodes = null;
            try {
                childrenNodes = registryClient.getChildrenKeys(workerGroupPath);
            } catch (Exception e) {
                log.error("getChildrenNodes exception: {}, workerGroupPath: {}", e.getMessage(), workerGroupPath);
            }
            if (childrenNodes == null || childrenNodes.isEmpty()) {
                continue;
            }
            WorkerGroup wg = new WorkerGroup();
            wg.setName(workerGroup);
            if (isPaging) {
                wg.setAddrList(String.join(Constants.COMMA, childrenNodes));
                String registeredValue = registryClient.get(workerGroupPath + Constants.SINGLE_SLASH + childrenNodes.iterator().next());
//                String[] rv = registeredValue.split(Constants.COMMA);
                String[] rv = registeredValue.split("@@");
                wg.setCreateTime(new Date(Long.parseLong(rv[6])));
                wg.setUpdateTime(new Date(Long.parseLong(rv[7])));
                wg.setSystemDefault(true);
            }
            workerGroups.add(wg);
        }
        return workerGroups;
    }

    /**
     * delete worker group by id
     *
     * @param id worker group id
     * @return delete result code
     */
    @Override
    @Transactional
    public Map<String, Object> deleteWorkerGroupById(User loginUser, Integer id) {
        Map<String, Object> result = new HashMap<>();
//        if (!canOperatorPermissions(loginUser, null, AuthorizationType.WORKER_GROUP, WORKER_GROUP_DELETE)) {
//            putMsg(result, Status.USER_NO_OPERATION_PERM);
//            return result;
//        }
        WorkerGroup workerGroup = workerGroupMapper.selectById(id);
        if (workerGroup == null) {
            log.error("Worker group does not exist, workerGroupId:{}.", id);
            putMsg(result, Status.DELETE_WORKER_GROUP_NOT_EXIST);
            return result;
        }
        // 查询当前worker分组是否是其他分组的关联节点
        LambdaQueryWrapper<WorkerGroup> lqw = Wrappers.lambdaQuery();
        lqw.apply("FIND_IN_SET({0}, connected_list)", workerGroup.getName());
        List<WorkerGroup> workerGroupList = workerGroupMapper.selectList(lqw);
        if (CollectionUtils.isNotEmpty(workerGroupList)) {
            List<String> workerGroupNames =
                    workerGroupList.stream().map(WorkerGroup::getName).collect(Collectors.toList());
            log.warn(
                    "Delete worker group failed because there is worker_group connect it, workerGroupList:{}.", workerGroupNames);
            putMsg(result, Status.DELETE_WORKER_GROUP_BY_CONNECT, workerGroupNames);
            return result;
        }
        List<ProcessInstance> processInstances = processInstanceMapper
                .queryByWorkerGroupNameAndStatus(workerGroup.getName(),
                        org.apache.dolphinscheduler.service.utils.Constants.NOT_TERMINATED_STATES);
        if (CollectionUtils.isNotEmpty(processInstances)) {
            List<Integer> processInstanceIds =
                    processInstances.stream().map(ProcessInstance::getId).collect(Collectors.toList());
            log.warn(
                    "Delete worker group failed because there are {} processInstances are using it, processInstanceIds:{}.",
                    processInstances.size(), processInstanceIds);
            putMsg(result, Status.DELETE_WORKER_GROUP_BY_ID_FAIL, processInstances.size());
            return result;
        }
        List<EnvironmentWorkerGroupRelation> environmentWorkerGroupRelationList =
                environmentWorkerGroupRelationMapper.queryByWorkerGroupName(workerGroup.getName());
        if (CollectionUtils.isNotEmpty(environmentWorkerGroupRelationList)) {
            putMsg(result, Status.DELETE_WORKER_GROUP_BY_ID_FAIL_ENV, environmentWorkerGroupRelationList.size(),
                    workerGroup.getName());
            return result;
        }
        workerGroupMapper.deleteById(id);
        processInstanceMapper.updateProcessInstanceByWorkerGroupName(workerGroup.getName(), "");
        log.info("Delete worker group complete, workerGroupName:{}.", workerGroup.getName());
        userGroupWorkerService.deleteByWorkId(id);
        putMsg(result, Status.SUCCESS);
        return result;
    }

    /**
     * query all worker address list
     *
     * @return all worker address list
     */
    @Override
    public Map<String, Object> getWorkerAddressList() {
        Map<String, Object> result = new HashMap<>();
        Set<String> serverNodeList = registryClient.getServerNodeSet(RegistryNodeType.WORKER);
        result.put(Constants.DATA_LIST, serverNodeList);
        putMsg(result, Status.SUCCESS);
        return result;
    }

    @Override
    public String getTaskWorkerGroup(TaskInstance taskInstance) {
        if (taskInstance == null) {
            return null;
        }

        String workerGroup = taskInstance.getWorkerGroup();

        if (StringUtils.isNotEmpty(workerGroup)) {
            return workerGroup;
        }
        int processInstanceId = taskInstance.getProcessInstanceId();
        ProcessInstance processInstance = processService.findProcessInstanceById(processInstanceId);

        if (processInstance != null) {
            return processInstance.getWorkerGroup();
        }
        log.info("task : {} will use default worker group", taskInstance.getId());
        return Constants.DEFAULT_WORKER_GROUP;
    }

    @Override
    public Map<Long, String> queryWorkerGroupByProcessDefinitionCodes(List<Long> processDefinitionCodeList) {
        List<Schedule> processDefinitionScheduleList =
                scheduleMapper.querySchedulesByProcessDefinitionCodes(processDefinitionCodeList);
        return processDefinitionScheduleList.stream().collect(Collectors.toMap(Schedule::getProcessDefinitionCode,
                Schedule::getWorkerGroup));
    }

    @Override
    public Map<String, MonitorDto> queryWorkerMonitorData() {
        List<WorkerServerModel> workerServers = monitorService.getServerListFromRegistry(false)
                .stream()
                .map((Server server) -> {
                    WorkerServerModel model = new WorkerServerModel();
                    model.setId(server.getId());
                    model.setHost(server.getHost());
                    model.setPort(server.getPort());
                    model.setZkDirectories(Sets.newHashSet(server.getZkDirectory()));
                    model.setResInfo(server.getResInfo());
                    model.setCreateTime(server.getCreateTime());
                    model.setLastHeartbeatTime(server.getLastHeartbeatTime());
                    return model;
                })
                .collect(Collectors.toList());
        Map<String, WorkerServerModel> workerHostPortServerMapping = workerServers
                .stream()
                .collect(Collectors.toMap(
                        (WorkerServerModel worker) -> {
                            String[] s = worker.getZkDirectories().iterator().next().split("/");
                            return s[s.length - 1];
                        }, Function.identity(), (WorkerServerModel oldOne, WorkerServerModel newOne) -> {
                            oldOne.getZkDirectories().addAll(newOne.getZkDirectories());
                            return oldOne;
                        }));

        Map<String, MonitorDto> WorkerMonitorDataMap = new HashMap<>();
        for (Map.Entry<String, WorkerServerModel> stringWorkerServerModelEntry : workerHostPortServerMapping
                .entrySet()) {
            WorkerServerModel value = stringWorkerServerModelEntry.getValue();
            MonitorDto dto = JSONUtils.parseObject(value.getResInfo(), MonitorDto.class);
            if (Objects.nonNull(dto)) {
                dto.setCreateTime(value.getCreateTime());
                dto.setPort(value.getPort());
            }
            WorkerMonitorDataMap.put(stringWorkerServerModelEntry.getKey(), dto);
        }
        return WorkerMonitorDataMap;
    }

     @Override
    public WorkerGroup getWorkerGroupById(Integer groupId) {
        return workerGroupMapper.selectById(groupId);
    }
     
     public WorkerGroup queryWorkGroupByIp(String ip) {
    	 QueryWrapper<WorkerGroup> qw = new QueryWrapper<WorkerGroup>();
    	 qw.like("addr_list", "%" + ip + "%");
    	 qw.orderByDesc("create_time");
    	 qw.last("LIMIT 1");
    	 return workerGroupMapper.selectOne(qw);
     }
     
     private List<WorkerGroup> getWorkGroupByWordIds(Integer userGroupId, List<WorkerGroup> lists){
    	 List<Integer> workIds = userGroupWorkerService.getWorkIdList(userGroupId);
    	 if(workIds != null && workIds.size() > 0) {
    		 List<WorkerGroup> wgList = new ArrayList<WorkerGroup>();
    		 for(Integer workId : workIds) {
    			 for(WorkerGroup wg : lists) {
    				 if(wg.getId() == workId) {
    					 wgList.add(wg);
    				 }
    			 }
    		 }
    		 return wgList;
    	 } else {
    		 return lists;
    	 }
     }
}
